package com.github.binarylei.hadoop.flowsum;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * 用来描述一个特定的作业
 * 比如，该作业使用哪个类作为逻辑处理中的map，哪个作为reduce
 * 还可以指定该作业要处理的数据所在的路径
 * 还可以指定改作业输出的结果放到哪个路径
 */
public class FlowSumRunner extends Configured implements Tool {

    @Override
    public int run(String[] args) throws Exception {
        //1. 配制文件
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        //2. 设置整个job所用的那些类在哪个jar包
        job.setJarByClass(FlowSumRunner.class);

        //3. 本job使用的mapper和reducer的类
        job.setMapperClass(FlowSumMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        //4. 指定reduce的输出数据kv类型
        job.setReducerClass(FlowSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        //5. 指定要处理的输入和输出数据存放路径
        FileInputFormat.setInputPaths(job, new Path("d://hadoop/flow/input/"));
        FileOutputFormat.setOutputPath(job, new Path("d://hadoop/flow/output/"));

        //6. 将job提交给集群运行
        return job.waitForCompletion(true) ? 0 : 1;
    }
    
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new FlowSumRunner(), args);
        System.exit(res);
    }
}
